package com.boot.study.table

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors._

object TableApiTest {
  def main(args: Array[String]): Unit = {
    // 1: 创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    //    // 1.1 基于老版本planner的流处理
    //    val settings: EnvironmentSettings = EnvironmentSettings
    //      .newInstance()
    //      .useOldPlanner()
    //      .inStreamingMode()
    //      .build()
    //    val oldStreamTableEnv = StreamTableEnvironment.create(env, settings)
    //
    //    // 1.2 基于老版本planner的批处理
    //    val batchEnv: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    //    val oldBatchTabelEnv = BatchTableEnvironment.create(batchEnv)
    //
    //    // 1.3 基于blink planner的流处理
    //    val blinkStreamSettings: EnvironmentSettings = EnvironmentSettings
    //      .newInstance()
    //      .useBlinkPlanner()
    //      .inStreamingMode()
    //      .build()
    //    val blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings)
    //
    //    // 1.4 基于blink planner的批处理
    //    val blinkBatchSettings: EnvironmentSettings = EnvironmentSettings
    //      .newInstance()
    //      .useBlinkPlanner()
    //      .inBatchMode()
    //      .build()
    //    val blinkBatchTableEnv = TableEnvironment.create(blinkBatchSettings)
    // 2. 连接外部系统，读取数据，注册表
    // 2.1 读取文件
    val filePath = "D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt"
    tableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("inputTable")
    //    val inputTable: Table = tableEnv.from("inputTable")
    //    inputTable.toAppendStream[(String, Long, Double)].print()

    // windows执行命令 kafka-console-producer.bat --broker-list 127.0.0.1:9092 --topic sensor
    // 2 从kafka读取数据
    //    tableEnv.connect(new Kafka()
    //      .version("0.11")
    //      .topic("sensor")
    //      .property("zookeeper.connect", "localhost:2181")
    //      .property("bootstrap.servers", "localhost:9092")
    //    )
    //      .withFormat(new Csv())
    //      .withSchema(new Schema()
    //        .field("id", DataTypes.STRING())
    //        .field("timestamp", DataTypes.BIGINT())
    //        .field("temperature", DataTypes.DOUBLE())
    //      )
    //      .createTemporaryTable("kafkaInputTable")
    //    val inputTable: Table = tableEnv.from("kafkaInputTable")
    //    inputTable.toAppendStream[(String, Long, Double)].print()

    // 3 查询转换
    // 3.1 使用table api
    val sensorTable: Table = tableEnv.from("inputTable")
    sensorTable
      .select('id, 'temperature)
      .filter('id === "Sensor_1")

    // 3.2 SQL实现
    val resultSqlTable: Table = tableEnv.sqlQuery(
      """
        |select id,temperature
        |from inputTable
        |where id = 'Sensor_1'
        |""".stripMargin)
    resultSqlTable.toAppendStream[(String,Double)].print("sql")

    env.execute("table api test")
  }
}
